// 线程局部存储实现高效广播的核心思想
// 按线程分治连接集合: 
// - 利用 ThreadLocalSingleton<ConnectionList>，让每个工作线程维护属于自己的连接列表。这样广播时只访问本线程的数据，避免跨线程共享容器所带来的锁竞争。
// 
// 消息分发采用函数投递:
// - 收到消息后，在触发线程中构造一个任务（Functor），然后通过 EventLoop::queueInLoop 投递到所有 EventLoop。每个线程收到任务后在自身上下文中遍历本地连接并发送消息。
// 
// 最小化同步开销:
// - 共享数据只剩一个 loops_ 集合，用来记录所有 EventLoop 指针，在插入或遍历时用互斥量保护。连接列表和消息发送完全在各自线程中完成，大幅降低锁使用频率，实现高性能转发。

#include "codec.h"

#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/base/ThreadLocalSingleton.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include <set>
#include <stdio.h>
#include <unistd.h>

using namespace muduo;
using namespace muduo::net;

// ChatServer 的职责：在多线程环境下维护聊天室连接集合，并把消息高效广播到每个线程拥有的连接上。
// 核心思想：
//
// 1. 每个 worker 线程（即每个 EventLoop）使用 ThreadLocalSingleton<ConnectionList> 存储自己的连接集合，
//    避免多个线程竞争同一份共享数据，从而减少锁竞争，提高转发效率。
// 2. 当收到一条消息时，主线程遍历所有 EventLoop，把“分发消息”的任务以 Functor 的形式投递到目标线程的消息队列（queueInLoop）。
//    这样每个线程在自己的上下文中给本线程管理的连接广播消息，完全无需跨线程访问连接集合。
// 3. loops_ 保存所有 EventLoop 指针，用于消息广播时遍历。该集合需要互斥量保护，因为多个线程会向其中插入（在 threadInit 中）。
class ChatServer : noncopyable
{
 public:
  ChatServer(EventLoop* loop,
             const InetAddress& listenAddr)
  : server_(loop, listenAddr, "ChatServer"),
    // LengthHeaderCodec 封装了长度前缀编解码逻辑。这里把 ChatServer::onStringMessage 注册为业务回调。
    codec_(std::bind(&ChatServer::onStringMessage, this, _1, _2, _3))
  {
    // TcpServer 的连接回调：当连接建立/断开时调用 ChatServer::onConnection
    server_.setConnectionCallback(
        std::bind(&ChatServer::onConnection, this, _1));
    // TcpServer 的消息回调：先交给 codec_ 解码，再回调到 ChatServer::onStringMessage
    server_.setMessageCallback(
        std::bind(&LengthHeaderCodec::onMessage, &codec_, _1, _2, _3));
  }

  // 设置底层线程池的线程数量（包括 I/O 线程 + 多个 worker 线程）
  void setThreadNum(int numThreads)
  {
    server_.setThreadNum(numThreads);
  }

  // 启动服务器前，先设置每个线程的初始化回调（threadInit），确保 thread local 容器正确构造
  void start()
  {
    server_.setThreadInitCallback(std::bind(&ChatServer::threadInit, this, _1));
    server_.start();
  }

 private:
  // onConnection 在对应连接所属的 EventLoop 线程中执行
  void onConnection(const TcpConnectionPtr& conn)
  {
    LOG_INFO << conn->peerAddress().toIpPort() << " -> "
             << conn->localAddress().toIpPort() << " is "
             << (conn->connected() ? "UP" : "DOWN");

    if (conn->connected())
    {
      // LocalConnections::instance() 取得当前线程的 thread local 容器
      // 每个线程都有自己独立的 ConnectionList，当连接归属该线程时只加入到本线程的集合里。
      LocalConnections::instance().insert(conn);
    }
    else
    {
      // 连接断开，从当前线程的集合中删除即可；不会影响其他线程。
      LocalConnections::instance().erase(conn);
    }
  }

  // onStringMessage 在消息所在连接的线程中触发，负责把消息广播任务转发给每个 EventLoop
  void onStringMessage(const TcpConnectionPtr&,
                       const string& message,
                       Timestamp)
  {
    // Functor f 封装了“在目标线程分发消息”的操作，绑定当前 ChatServer 实例和消息内容。
    EventLoop::Functor f = std::bind(&ChatServer::distributeMessage, this, message);
    LOG_DEBUG;

    // 遍历所有 EventLoop 时需要保护 loops_；多个线程可能会同时访问该集合（例如 threadInit 中插入新 loop）。
    MutexLockGuard lock(mutex_);
    for (std::set<EventLoop*>::iterator it = loops_.begin();
        it != loops_.end();
        ++it)
    {
      // queueInLoop 把 Functor 投递到对应 EventLoop 的任务队列中，
      // 由该 EventLoop 在线程上下文里执行 distributeMessage，从而在本线程内广播。
      (*it)->queueInLoop(f);
    }
    LOG_DEBUG;
  }

  typedef std::set<TcpConnectionPtr> ConnectionList;

  // distributeMessage 在各个 EventLoop 的线程上下文中执行，访问的是 thread local 的连接集合
  void distributeMessage(const string& message)
  {
    LOG_DEBUG << "begin";
    // LocalConnections::instance() 始终返回“当前线程”所对应的 ConnectionList
    // 因为每个线程各自持有一份副本，所以这里无需加锁。
    for (ConnectionList::iterator it = LocalConnections::instance().begin();
        it != LocalConnections::instance().end();
        ++it)
    {
      // codec_.send 内部会把 message 加上长度头并调用 TcpConnection::send
      codec_.send(get_pointer(*it), message);
    }
    LOG_DEBUG << "end";
  }

  // threadInit 在每个 worker 线程启动时被调用一次
  void threadInit(EventLoop* loop)
  {
    // ThreadLocalSingleton::pointer() 在 thread local 变量未初始化时返回 NULL
    assert(LocalConnections::pointer() == NULL);
    // 通过 instance() 创建并持有当前线程的 ConnectionList
    LocalConnections::instance();
    assert(LocalConnections::pointer() != NULL);
    // loops_ 中记录所有 EventLoop 指针，用于广播消息时遍历
    MutexLockGuard lock(mutex_);
    loops_.insert(loop);
  }

  TcpServer server_;
  LengthHeaderCodec codec_;
  typedef ThreadLocalSingleton<ConnectionList> LocalConnections;

  // mutex_ 用于保护跨线程共享的 loops_ 集合
  MutexLock mutex_;
  std::set<EventLoop*> loops_ GUARDED_BY(mutex_);
};

int main(int argc, char* argv[])
{
  LOG_INFO << "pid = " << getpid();
  if (argc > 1)
  {
    EventLoop loop;
    uint16_t port = static_cast<uint16_t>(atoi(argv[1]));
    InetAddress serverAddr(port);
    ChatServer server(&loop, serverAddr);
    if (argc > 2)
    {
      server.setThreadNum(atoi(argv[2]));
    }
    server.start();
    loop.loop();
  }
  else
  {
    printf("Usage: %s port [thread_num]\n", argv[0]);
  }
}